package com.atguigu.edu.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSONObject;

import com.atguigu.edu.util.DimUtil;
import com.atguigu.edu.util.DruidDSUtil;
import com.atguigu.edu.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;


import java.sql.Connection;
import java.sql.SQLException;
import java.util.Collections;
import java.util.concurrent.ExecutorService;


public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimJoinFunction<T> {

    private ExecutorService executorService;
    private DruidDataSource dataSource;
    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        executorService = ThreadPoolUtil.getInstance();
        dataSource = DruidDSUtil.createDataSource();
    }

    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        // 开启多个线程，发送异步请求
        executorService.submit(
                new Runnable() {
                    @Override
                    public void run() {
                        Connection conn = null;
                        try {
                            conn = dataSource.getConnection();
                            //根据流中的对象 获取要关联的维度的主键
                            String key = getKey(obj);
                            //根据维度的主键获取对应的维度对象
                            JSONObject dimInfoJsonObj = DimUtil.getDimInfo(conn, tableName, key);
                            if(dimInfoJsonObj != null){
                                //将维度对象的属性补充到流中对象上
                                join(obj,dimInfoJsonObj);
                            }
                            //获取数据库交互的结果并发送给ResultFuture的回调函数
                            resultFuture.complete(Collections.singleton(obj));
                        } catch (Exception e) {
                            e.printStackTrace();
                            System.out.println("~~发送异步请求进行维度关联发生了异常~~");
                        } finally {
                            if(conn != null){
                                try {
                                    conn.close();
                                } catch (SQLException e) {
                                    e.printStackTrace();
                                }
                            }
                        }

                    }
                }
        );
    }
}